package com.icehand.canal.client;

import com.alibaba.google.common.util.concurrent.ThreadFactoryBuilder;
import com.alibaba.otter.canal.client.CanalConnector;
import com.icehand.canal.client.transfer.TransponderFactory;
import com.icehand.canal.config.CanalConfig;
import com.icehand.canal.handler.HandlerWrapper;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author icehand
 */
@Slf4j
public class SimpleCanalClient extends BaseCanalClient {
    private ThreadPoolExecutor executor;
    private final HandlerWrapper wrapper;
    public SimpleCanalClient(CanalConfig config, TransponderFactory factory, HandlerWrapper handlerWrapper){
        super(config,factory);
        ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()
                .setNameFormat("simple-canal-client-pool-%d").build();
        executor = new ThreadPoolExecutor(0,Integer.MAX_VALUE, 60L, TimeUnit.SECONDS,
                new SynchronousQueue<>(), namedThreadFactory);
        this.wrapper = handlerWrapper;

    }

    @Override
    protected void process(CanalConnector connector, CanalConfig.Instance instance,String key) {
        executor.submit(factory.newTransponder(connector,instance,key,wrapper.getHandlers(),wrapper.getPoints()));
    }

    @Override
    public void stop() {
        super.stop();
        executor.shutdown();
    }

    @Override
    public boolean isRunning() {
        return false;
    }
}
